/*=====================================================================

  File:      FetchAsync.cs
  Summary:   SQL Remote Blob Storage sample application - FetchAsync.
             Sample application used to fetch blob data from the 
             sample application's tables. Demonstrates using the 
             System.IO.Stream asynchronous methods with RBS.
  Date:	     July 22st, 2008

---------------------------------------------------------------------

  This file is part of the Microsoft SQL Server Code Samples.
  Copyright (C) Microsoft Corporation.  All rights reserved.

This source code is intended only as a supplement to Microsoft
Development Tools and/or on-line documentation.  See these other
materials for detailed information regarding Microsoft code samples.

THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY
KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
PARTICULAR PURPOSE.

======================================================= */

using System;
using System.Data;
using System.Data.SqlClient;
using System.Globalization;
using System.IO;
using System.Threading;
using System.Text;

using Microsoft.Data.SqlRemoteBlobs;

namespace Microsoft.Samples.RemoteBlobStorage
{
    /// <summary>
    /// Sample Remote Blob Storage Application.
    /// </summary>
    public
    sealed
    class FetchAsync
    {
        private FetchAsync() { }

        public
        static
        void RunFetchAsync(string connectionString)
        {
            Console.WriteLine("Running the FetchAsync Sample");

            // Reads all blob data from the application's tables.
            //
            using (SqlConnection conn = new SqlConnection(connectionString))
            {
                conn.Open();

                // Create a blob context to operate againt RBS.
                //
                SqlRemoteBlobContext blobContext = new SqlRemoteBlobContext(conn);

                // Fetch this blob using the blobId retrieved from running a query against the SearchTags field.
                // No specific collection information is required for Fetch as this information is coded into
                // the BlobId.
                //
                // Using the rbs_fn_get_blob_reference function converts the BlobId into a blob reference, which 
                // contains store specific information regarding the location of the blob. This is an optional step - 
                // passing the blob id to the SqlRemoteBlobContext.OpenBlob method will still function but will 
                // require an extra internal round trip to the database to do the conversion.
                //
                string queryBlobEntry = @" 
                                select 
                                    mssqlrbs.rbs_fn_get_blob_reference(BlobId)
                                from 
                                    ApplicationSampleTable 
                                where 
                                    SearchTags LIKE '%Sample%'
                                ";

                DataTable blobReferences = new DataTable();
                blobReferences.Locale = CultureInfo.InvariantCulture;

                // Run a query, returning all the blob IDs.
                //
                using (SqlCommand cmd = new SqlCommand(queryBlobEntry, conn))
                {
                    // Fill the blob references and close the command.
                    //
                    using (SqlDataAdapter adapter = new SqlDataAdapter(cmd))
                    {
                        adapter.Fill(blobReferences);
                    }
                }

                // There should be no further database access required from here on.
                // Close the database connection to confirm this is the case.
                //
                conn.Close();

                // Cycle through all the returned rows and retrieve all the blob references.
                // Store these in a data structure and process individually.
                //
                byte[][] blobReferenceArray = new byte[blobReferences.Rows.Count][];

                for (int i = 0; i < blobReferences.Rows.Count; i++)
                {
                    // Get the BlobReference
                    //
                    blobReferenceArray[i] = (byte[])blobReferences.Rows[i][0];
                }

                // Process the array of blob references.
                //
                Console.WriteLine("Beginning asynchronous processing:");
                ProcessBlobReferences(blobContext, blobReferenceArray);
            }

            // Success
            //
        }

        /// <summary>
        /// Threads must decrement s_BlobsToProcess protecting
        /// their access with a mutex.
        /// </summary>
        public static int s_BlobsToProcess;
        public static Object BlobsToProcessMutex = new Object[0];
        public static AsyncCallback s_BlobStreamCallback;

        /// <summary>
        /// This is the buffer size that will be stored in each state object. 
        /// Once this buffer is full, the callback will occur. For this demo the 
        /// buffer size is small, however for a production system this should be
        /// a compromise between minimizing local memory pressure and ensuring
        /// each asynchronous call has a significant amount of work to do.
        /// </summary>
        public const int c_BufferSize = 70;

        /// <summary>
        /// WaitObject is signalled when all blob processing is done.
        /// </summary>
        public static Object WaitObject = new Object[0];

        /// <summary>
        /// This object is used to keep track of in-flight blob operations.
        /// </summary>
        public class BlobStateObject
        {
            public byte[] blobReference;
            public byte[] dataBuffer = new byte[c_BufferSize];
            public int blobNumber;
            public SqlRemoteBlob blob;
        }

        /// <summary>
        /// Callback made when BeginRead has filled the data buffer with blob data.
        /// Determines if the current blob has finished processing and either fires 
        /// off another asynchronous read or completes this blob's processing.
        ///
        /// For this application partial blob information is written to the console.
        /// On a production system this is where the blob information should be 
        /// processed, for instance output to a file or perform some data transform.
        /// </summary>
        public 
        static 
        void BlobStreamCallback(IAsyncResult asyncResult)
        {
            BlobStateObject state = (BlobStateObject)asyncResult.AsyncState;

            int bytesRead = state.blob.EndRead(asyncResult);

            // Blob read is not complete.
            //
            if (bytesRead > 0)
            {
                Console.WriteLine(string.Format(CultureInfo.CurrentCulture,
                    "BlobNumber {0}: {1}", state.blobNumber, 
                    Encoding.Unicode.GetString(state.dataBuffer, 0, bytesRead)));
                
                // Read the next buffer worth of data from the stream.
                //
                state.blob.BeginRead(state.dataBuffer,
                                     0,
                                     state.dataBuffer.Length,
                                     s_BlobStreamCallback,
                                     state);
            }
            else
            {
                // Blob read is complete.
                //
                state.blob.Close();
                state.dataBuffer = null;
                
                // Record that this blob is finished now.
                //
                lock (BlobsToProcessMutex)
                {
                    s_BlobsToProcess--;
                    if (s_BlobsToProcess == 0)
                    {
                        Monitor.Enter(WaitObject);
                        Monitor.Pulse(WaitObject);
                        Monitor.Exit(WaitObject);
                    }
                }
            }
        }

        /// <summary>
        /// Processes each blob reference in turn and fires off the BeginRead asynchronous 
        /// System.IO.Stream call.
        /// </summary>
        public
        static
        void ProcessBlobReferences(SqlRemoteBlobContext context, byte[][] blobReferences)
        {
            Console.WriteLine("Processing blob references. Using buffer size {0}.", c_BufferSize);

            long t0 = Environment.TickCount;

            int totalBlobCount = blobReferences.GetLength(0);
            s_BlobsToProcess = totalBlobCount;

            s_BlobStreamCallback = new AsyncCallback(BlobStreamCallback);

            for (int i = 0; i < totalBlobCount; i++)
            {
                BlobStateObject state = new BlobStateObject();
                state.blobReference = blobReferences[i];
                state.blobNumber = i;

                // Create the SqlRemoteBlob object using the SqlRemoteBlobContext.
                //
                state.blob = context.OpenBlob(state.blobReference);

                // Begin reading from the stream.
                //
                state.blob.BeginRead(state.dataBuffer,
                                     0,
                                     state.dataBuffer.Length,
                                     s_BlobStreamCallback,
                                     state);
            }

            // Determine whether all blobs are done being processed.  
            // If not, block until all are finished.
            //
            bool mustBlock = false;
            lock (BlobsToProcessMutex)
            {
                if (s_BlobsToProcess > 0)
                    mustBlock = true;
            }
            if (mustBlock)
            {
                Console.WriteLine("All worker threads are queued. Blobs to process: {0}.", s_BlobsToProcess);
                Monitor.Enter(WaitObject);
                Monitor.Wait(WaitObject);
                Monitor.Exit(WaitObject);
            }

            long t1 = Environment.TickCount;
            Console.WriteLine("Total time reading blobs: {0} ms", t1 - t0);
        }
    }
}
